package com.li.flink.app

import com.alibaba.fastjson.JSON
import com.li.flink.domain.Access
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object OsUserCntV2 {
  def main(args: Array[String]): Unit = {
    val inputPath = "/Users/lijiacen/Downloads/access.json"
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val textStream = env.readTextFile(inputPath)
    val mapStream: DataStream[Access] = textStream.map(m => {
      try {
        val obj: Access = JSON.parseObject(m, classOf[Access])
        obj
      } catch {
        case ex: Exception => print("json数据解析失败：" + m)
          null
      }
    })
    val filterStream = mapStream.filter(m => {
      m.getEvent.equals("startup")
    }).map(m => {
      (m.getNu, 1)
    })
    val keySelector = new KeySelector[Tuple2[Int, Int], Int] {
      override def getKey(in: Tuple2[Int, Int]): Int = {
        in._1
      }
    }
    //操作系统和新老用户维度
    val keyByStream = filterStream.keyBy(keySelector).sum(1).print()
    env.execute("OsUserCntV2")
  }
}
